const prefix = '/api/deployment'
const spawn = require('cross-spawn')
const { processTemplate } = require('../utils/str')
const { fullPath, basename, statSync, createZip, readFileSync } = require('../utils/file')
const { randomUUID } = require('crypto')
const BlockingQueue = require('sync').BlockingQueue
const axios = require('../axios/index')
const FormData = require('form-data')
const pLimit = require('p-limit')
const logOutputQueueMap = new Map()
const childProcessorMap = new Map()
const abortControllerMap = new Map()
const STREAM_DONE = '[DONE]' // 流式传输结束的标识
const CHUNK_SIZE = 1024 * 1024 // 每个分片1MB
const CHUNKS_UPLOAD_CONCURRENCY = 5 // 限制分片上传并发数为5

function run(Joi) {
  this.method = 'post'
  this.path = '/run'
  this.validate = {
    type: 'json',
    body: {
      project: Joi.string().optional().allow(''),
      module: Joi.string().optional().allow(''),
      build: Joi.string().optional().allow(''),
      filePath: Joi.string().required(),
      uploadTo: Joi.string().required(),
      ssh: Joi.string().optional().allow(''),
      shell: Joi.string().optional().allow('')
    }
  }
  
  return async ctx => { 
    const {
      project,
      module,
      build,
      filePath,
      uploadTo,
      ssh,
      shell
    } = ctx.request.body
    const ctxId = randomUUID()
    const logOutputQueue = new BlockingQueue()
    logOutputQueueMap.set(ctxId, logOutputQueue)
    new Promise((resolve, reject) => {
      if (!project || !build) {
        resolve(0)
        return
      }

      const [ command, ...args ] = processTemplate(build, { module }).split(/\s+/)
      const childProcessor = spawn(command, args, { cwd: project })
      childProcessorMap.set(ctxId, childProcessor)

      // 监听标准输出
      childProcessor.stdout.on('data', data => {
        logOutputQueue.add(data)
      })

      // 监听标准错误输出
      childProcessor.stderr.on('data', data => {
        logOutputQueue.add(data)
      })

      // 监听进程关闭事件
      childProcessor.on('close', resolve)

      // 监听错误事件
      childProcessor.on('error', reject)
    })
    .then(code => {
      if (code !== 0) {
        return Promise.reject(new Error(`Exit value: ${code}`))
      }
      
      const fullFilePath = fullPath(project, processTemplate(filePath, { module }))
      const executableShell = processTemplate(shell, { module, uploadTo, filename: basename(fullFilePath) })
      return uploadAndDeploy(
        fullFilePath, 
        uploadTo, 
        ssh, 
        executableShell, 
        ctxId
      )
    })
    .catch(err => {
      console.error(err)
      logOutputQueue.add(err)
    })
    .finally(() => {
      logOutputQueue.add(STREAM_DONE)
      childProcessorMap.delete(ctxId)
      abortControllerMap.delete(ctxId)
    })

    ctx.status = 201
    ctx.body = ctxId
  }
}

async function uploadAndDeploy(filePath, uploadTo, ssh, shell, ctxId) {
  if (statSync(filePath).isDirectory()) {
    filePath = await createZip(filePath)
  }

  const abortController = new AbortController()
  abortControllerMap.set(ctxId, abortController)

  const formData = new FormData()
  formData.append('uploadTo', uploadTo)
  formData.append('ssh', ssh)
  formData.append('shell', shell)
  const fileBuffer = readFileSync(filePath)
  const filename = basename(filePath)
  if (fileBuffer.length < CHUNK_SIZE) {
    formData.append('file', fileBuffer, filename)
  } else {
    const remoteFilePath = await uploadChunks(fileBuffer, filename, uploadTo, ctxId)
    formData.append('filePath', remoteFilePath)
  }
  
  return new Promise((resolve, reject) => {
    axios.post('/api/exec/deploy', formData, {
      headers: { 'X-Request-ID': ctxId },
      responseType: 'stream',
      signal: abortController.signal
    })
    .then(stream => {
      const logOutputQueue = logOutputQueueMap.get(ctxId)
      stream.on('data', chunk => {
        const chunkStr = chunk.toString()
        if (chunkStr !== '\n\n') {
          let data = chunkStr
          if (data.startsWith('data:')) {
            data = data.substring(5)
          }
    
          logOutputQueue.add(data)
        }
      })
    
      stream.on('end', resolve)
    
      stream.on('error', reject)
    })
    .catch(reject)
  })
}

async function uploadChunks(fileBuffer, filename, uploadTo, ctxId) {
  const totalChunks = Math.ceil(fileBuffer.length / CHUNK_SIZE)
  const logOutputQueue = logOutputQueueMap.get(ctxId)
  const abortController = abortControllerMap.get(ctxId)

  // 初始化分片上传
  logOutputQueue.add(`========================Upload chunks========================`)
  const uploadId = await axios.post('/api/upload/init', {
    uploadTo,
    filename,
    totalChunks
  }, {
    signal: abortController.signal
  })
  
  // 开始分片并上传
  const uploadPromises = []
  const limit = pLimit(CHUNKS_UPLOAD_CONCURRENCY)
  for (let i = 0; i < totalChunks; i++) {
    const start = i * CHUNK_SIZE
    const end = Math.min(start + CHUNK_SIZE, fileBuffer.length)
    const chunk = fileBuffer.subarray(start, end)
    const p = limit(() => {
      const formData = new FormData()
      formData.append('uploadId', uploadId)
      formData.append('chunkNum', i)
      formData.append('chunk', chunk, `chunk_${i}`)
      logOutputQueue.add(`========================Uploading chunk ${i + 1}/${totalChunks}========================`)
      return axios.post('/api/upload/chunk', formData, {
        signal: abortController.signal
      })
    })
    uploadPromises.push(p)
  }

  // 等待所有分片上传完成
  await Promise.all(uploadPromises)

  // 完成分片上传
  logOutputQueue.add(`========================Complete upload========================`)
  return axios.post(`/api/upload/${uploadId}/complete`, null, {
    signal: abortController.signal
  })
}

function streamingLogOutput(Joi) {
  this.method = 'get'
  this.path = '/:ctxId/log/stream'
  return async ctx => { 
    const { ctxId } = ctx.params
    const logOutputQueue = logOutputQueueMap.get(ctxId)
    if (logOutputQueue) {
      ctx.status = 200
      ctx.type = 'text/event-stream'
      await new Promise(async resolve => {
        while (true) {
          const chunk = await logOutputQueue.poll()
          ctx.res.write(`id: ${ctxId}\n\ndata: ${chunk}\n\n`)
          if (chunk === STREAM_DONE) {
            resolve()
          }
        }
      })
      logOutputQueueMap.delete(ctxId)
    }
    
    ctx.res.end()
  }
}

function cancel(Joi) {
  this.method = 'post'
  this.path = '/:ctxId/cancel'
  return async ctx => {
    const { ctxId } = ctx.params
    const logOutputQueue = logOutputQueueMap.get(ctxId)
    if (logOutputQueue) {
      logOutputQueue.add(STREAM_DONE)
    }

    setTimeout(() => {
      const childProcessor = childProcessorMap.get(ctxId)
      if (childProcessor) {
        childProcessor.kill()
      }

      const abortController = abortControllerMap.get(ctxId)
      if (abortController) {
        abortController.abort()
      }
    }, 0)
    ctx.status = 204
  }
}

module.exports = {
  prefix,
  run,
  streamingLogOutput,
  cancel
}